package com.huatai.bi.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.huatai.bi.dto.MqttClientDto;
import com.huatai.bi.entity.DirectoryEntity;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.mapper.MqMqttClientMapper;
import com.huatai.bi.mqtt.MqttClientCallback;
import com.huatai.bi.mqtt.MqttClientConnect;
import com.huatai.bi.service.DirectoryService;
import com.huatai.bi.service.MqMqttClientService;
import com.huatai.bi.utils.PaginationUtils;
import com.huatai.common.error.exception.ServiceException;
import jodd.util.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author songxiude
 * @description 针对表【mq_mqtt_client(mqtt链接信息表)】的数据库操作Service实现
 * @createDate 2024-07-15 14:36:47
 */
@Service
@Slf4j
public class MqMqttClientServicebkImpl extends ServiceImpl<MqMqttClientMapper, MqMqttClientEntity> {

	@Autowired
	private DirectoryService directoryService;

	/**
	 * 添加修改mqtt链接
	 *
	 * @param mqMqttClient
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public boolean mqttSave(MqMqttClientEntity mqMqttClient) {
		//判断链接是否已存在
//		long count = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, mqMqttClient.getClientId()));
//		if (ObjectUtils.isEmpty(mqMqttClient.getId()) && count > 0) {
//			throw new ServiceException("该链接已存在不可重复创建");
//		}
		//判断链接是否已经存在
		long count = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getServerHost, mqMqttClient.getServerHost())
			.eq(MqMqttClientEntity::getServerPort, mqMqttClient.getServerPort()));
		if (count > 0) {
			throw new ServiceException("该链接已存在不可重复创建");
		}
		//判断topic是否存在
		long tcount = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery().like(MqMqttClientEntity::getClientTopic, mqMqttClient.getClientTopic()));
		if (tcount > 0) {
			throw new ServiceException("所订阅的topic已经存在");
		}
		//判断数据集是不是目录
		DirectoryEntity directoryEntity = directoryService.getOne(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId, mqMqttClient.getDtId()));
		if (ObjectUtils.isEmpty(directoryEntity) || directoryEntity.getIsDir().intValue() != 0) {
			throw new ServiceException("数据集不是目录不能添加新的链接");
		}
		//组装链接地址前缀
		if (mqMqttClient.getClientType().intValue() == 1) {
			mqMqttClient.setMqttPrefix("tcp://");
		} else {
			mqMqttClient.setMqttPrefix("ssl://");
		}
		//如果不传客户端链接id自动生成
		if (StringUtil.isEmpty(mqMqttClient.getClientId())) {
			mqMqttClient.setClientId(IdWorker.get32UUID());
		}
		//链接mqtt
		mqMqttClient.setStatus(1);
		boolean mqB = this.initMqttServer(mqMqttClient);
		if (!mqB) {
			mqMqttClient.setStatus(0);
		}
		mqMqttClient.setEnableFlag(true);
		return save(mqMqttClient);
	}

	/**
	 * 修改mqtt链接信息
	 *
	 * @param mqMqttClient
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public boolean mqttUpdate(MqMqttClientEntity mqMqttClient) {
		//判断topic是否存在
		long tcount = this.count(Wrappers.<MqMqttClientEntity>lambdaQuery()
			.like(MqMqttClientEntity::getClientTopic, mqMqttClient.getClientTopic())
			.ne(MqMqttClientEntity::getId, mqMqttClient.getId()));
		if (tcount > 0) {
			throw new ServiceException("所订阅的topic已经存在");
		}
		//先停掉运行中的链接
		ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
		MqttClientConnect clientConnect = mqttClients.get(mqMqttClient.getClientId());
		if (ObjectUtil.isNotEmpty(clientConnect)) {
			try {
				clientConnect.close();
			} catch (MqttException e) {
				log.error("关闭链接失败");
				throw new ServiceException("关闭链接失败");
			}
		}
		mqttClients.remove(mqMqttClient.getClientId());
		//组装链接地址前缀
		if (mqMqttClient.getClientType().intValue() == 1) {
			mqMqttClient.setMqttPrefix("tcp://");
		} else {
			mqMqttClient.setMqttPrefix("ssl://");
		}
		//链接mqtt
		this.initMqttServer(mqMqttClient);
		mqMqttClient.setStatus(1);
		return updateById(mqMqttClient);
	}

	/**
	 * 获取mqtt链接信息
	 *
	 * @param mqttClientDto
	 * @return
	 */
	public IPage<MqMqttClientEntity> mqttList(MqttClientDto mqttClientDto) {
		Page<MqMqttClientEntity> page = PaginationUtils.<MqMqttClientEntity>getPage(mqttClientDto.getPage(), mqttClientDto.getSize());
		LambdaQueryWrapper<MqMqttClientEntity> wrapper = Wrappers.<MqMqttClientEntity>lambdaQuery();
		//排除部分敏感字段
		//wrapper.select(MqMqttClientEntity.class, client -> !client.getColumn().equals("password"));
		//wrapper.select(MqMqttClientEntity::getId,MqMqttClientEntity::getMqttName);
		wrapper.like(StringUtil.isNotEmpty(mqttClientDto.getMqttName()), MqMqttClientEntity::getMqttName, mqttClientDto.getMqttName());
		wrapper.orderByDesc(MqMqttClientEntity::getCreateTime);
		return page(page, wrapper);
	}

	/**
	 * 手动启动mqtt
	 *
	 * @param id
	 * @return
	 */
	public boolean startMqtt(Long id) {
		MqMqttClientEntity clientEntity = getById(id);
		if (ObjectUtils.isEmpty(clientEntity)) {
			throw new ServiceException("不存在的实例");
		}
		if (clientEntity.getStatus().intValue() == 1) {
			throw new ServiceException("已经正常链接请不要频繁重连");
		}
		boolean b = this.initMqttServer(clientEntity);
		if (b) {
			clientEntity.setStatus(1);
		}
		updateById(clientEntity);
		return b;
	}

	/**
	 * 删除mqtt链接
	 *
	 * @param clientId
	 * @return
	 */
	@Transactional(rollbackFor = Exception.class)
	public boolean removeMqtt(String clientId) {
		try {
			ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
			MqttClientConnect mqttClientConnect = mqttClients.get(clientId);
			if (ObjectUtil.isNotEmpty(mqttClientConnect)) {
				mqttClientConnect.close();
			}
			mqttClients.remove(clientId);
		} catch (MqttException e) {
			e.printStackTrace();
			log.error("断开连接失败,id--->" + clientId);
			//return false;
		}
		MqMqttClientEntity mqttClientEntity = getOne(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
		if (ObjectUtil.isEmpty(mqttClientEntity)) {
			throw new ServiceException("非法删除");
		}
		long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId, mqttClientEntity.getDataSetId()));
		if (count > 0) {
			throw new ServiceException("已存在数据集，请删除数据集");
		}
		return remove(Wrappers.<MqMqttClientEntity>lambdaQuery().eq(MqMqttClientEntity::getClientId, clientId));
	}

	/**
	 * 初始化链接mqtt
	 *
	 * @param mqMqttClient
	 */
	public boolean initMqttServer(MqMqttClientEntity mqMqttClient) {
		try {
			MqttClientConnect mqttClientConnect = new MqttClientConnect();
			mqttClientConnect.setMqttClientId(mqMqttClient.getClientId());
			mqttClientConnect.setMqttClient(mqMqttClient, new MqttClientCallback(mqMqttClient.getClientId()));
			mqttClientConnect.sub(StringUtils.commaDelimitedListToStringArray(mqMqttClient.getClientTopic()));
			MqttClientConnect.mqttClients.put(mqMqttClient.getClientId(), mqttClientConnect);
			log.info("mqtt初始化链接成功,host--->" + mqMqttClient.getServerHost());
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			log.error("新增连接失败,host--->" + mqMqttClient.getServerHost());
			return false;
		}
	}

	/**
	 * 停止Mqtt
	 *
	 * @param id
	 * @return
	 */
	public boolean stopMqtt(Long id) {
		MqMqttClientEntity clientEntity = getById(id);
		if (ObjectUtils.isEmpty(clientEntity)) {
			throw new ServiceException("不存在的链接实例");
		}
		//先停止mqtt链接
		ConcurrentHashMap<String, MqttClientConnect> mqttClients = MqttClientConnect.mqttClients;
		MqttClientConnect mqttClientConnect = mqttClients.get(clientEntity.getClientId());
		if (!ObjectUtils.isEmpty(mqttClientConnect)) {
			try {
				mqttClientConnect.close();
				mqttClients.remove(clientEntity.getClientId());
			} catch (MqttException e) {
				throw new ServiceException("停止mqtt链接失败");
			}
		}
		//更新状态
		clientEntity.setStatus(0);
		return updateById(clientEntity);
	}


}




